package ex.ex2;

import org.apache.storm.topology.BasicOutputCollector;
import org.apache.storm.topology.OutputFieldsDeclarer;
import org.apache.storm.topology.base.BaseBasicBolt;
import org.apache.storm.tuple.Fields;
import org.apache.storm.tuple.Tuple;
import org.apache.storm.tuple.Values;

import java.util.ArrayList;
import java.util.List;

/**
 * 判断是否包含关键字，是，就传递
 */
public class KeywordBoltA extends BaseBasicBolt {
    private final List<String> keywords = new ArrayList<String>(){
        {
            add("投诉");
            add("12315");
            add("12345");
            add("新冠");
            add("国家");
            add("有关部门");
        }
    };

    @Override
    public void execute(Tuple input, BasicOutputCollector collector) {
        String sentence = input.getString(1);
        for (String keyword : keywords) {
            if (sentence.contains(keyword)) {
                collector.emit(new Values(input.getValue(0), sentence));
                return;
            }
        }
    }

    @Override
    public void declareOutputFields(OutputFieldsDeclarer declarer) {
        declarer.declare(new Fields("id", "sentence"));
    }
}
